package org.atguigu.gmall.realtime.app.dim;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.atguigu.gmall.realtime.app.BaseApp;
import org.atguigu.gmall.realtime.bean.TableProcess;
import org.atguigu.gmall.realtime.common.Constant;
import org.atguigu.gmall.realtime.util.FlinkSinkUtil;
import org.atguigu.gmall.realtime.util.JdbcUtil;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Properties;

/**
 * Project:gmallRealTime
 * Package:org.atguigu.gmall.realtime.app.dim
 * Author: ZengHaiFeng
 * CreateTime:2023/4/19 16:31:01
 * Description: Todo
 * Version：1.0
 */
@Slf4j
public class DimApp extends BaseApp{
    public static void main(String[] args) {
        new DimApp().init(20001,2, Constant.TOPIC_ODS_DB,"DimApp");
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> ds) {
        SingleOutputStreamOperator<JSONObject> etlStream = etl(ds);
        SingleOutputStreamOperator<TableProcess> configTable = readTableProcess(env);
        configTable = createPhoenixTable(configTable);
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> dataStream = connect(etlStream, configTable);
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> resultStream = deleteNotNeedColumns(dataStream);
        writeToPhoenix(resultStream);
    }

    /**
     * @description: write final dim date to phoenix
     * @author: zenghaifeng
     * @date: 2023/4/22 11:12
     * @param resultStream
     **/
    private void writeToPhoenix(SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> resultStream) {
        resultStream.addSink(FlinkSinkUtil.getPhoenixSink());
    }

    /**
     * @description: delete not need columns
     * @author: zenghaifeng
     * @date: 2023/4/21 20:48
     * @param dataStream
     * @return SingleOutputStreamOperator<Tuple2<TableProcess>>
     **/
    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> deleteNotNeedColumns(
            SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> dataStream) {
        return dataStream.map(
                (MapFunction<Tuple2<JSONObject, TableProcess>, Tuple2<JSONObject, TableProcess>>) value -> {
                    // get data
                    JSONObject data = value.f0;
                    ArrayList<String> columns = new ArrayList<>(Arrays.asList(value.f1.getSinkColumns().split(",")));
                    columns.add("op_type");
                    data.keySet().removeIf(key -> !columns.contains(key));
//                    System.out.println(value);
                    return value;
                }).returns(Types.TUPLE(Types.GENERIC(JSONObject.class),Types.POJO(TableProcess.class)));
    }

    /**
     * @description: data stream connect to broadcast config stream.
     * @author: zenghaifeng
     * @date: 2023/4/21 20:14
     * @param etlStream
     * @param configTable
     * @return SingleOutputStreamOperator<Tuple2<TableProcess>>
     **/
    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> connect(
            SingleOutputStreamOperator<JSONObject> etlStream,
            SingleOutputStreamOperator<TableProcess> configTable){
        // convert a config stream to a broadcast stream
        MapStateDescriptor<String, TableProcess> configStateDescriptor = new MapStateDescriptor<>(
                "config",
                String.class,
                TableProcess.class);
        BroadcastStream<TableProcess> configStream = configTable.broadcast(configStateDescriptor);
        return etlStream.connect(configStream)
                .process(new BroadcastProcessFunction<JSONObject, TableProcess, Tuple2<JSONObject,TableProcess>>() {
                    @Override
                    public void processElement(JSONObject value,
                                               BroadcastProcessFunction<JSONObject, TableProcess, Tuple2<JSONObject, TableProcess>>.ReadOnlyContext ctx,
                                               Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {
                        String sourceTableName = value.getString("table");
                        String key = getKey(sourceTableName, "ALL");
                        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(configStateDescriptor);
                        TableProcess tableProcess = broadcastState.get(key);
                        if (tableProcess != null){
                            JSONObject data = value.getJSONObject("data");
                            data.put("op_type",value.getString("type"));
                            out.collect(Tuple2.of(data,tableProcess));
                        }
                    }

                    private String getKey(String sourceTableName, String type) {
                        return sourceTableName + ":" + type;
                    }

                    @Override
                    public void processBroadcastElement(TableProcess value,
                                                        BroadcastProcessFunction<JSONObject, TableProcess, Tuple2<JSONObject, TableProcess>>.Context ctx,
                                                        Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {
                        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(configStateDescriptor);
//                        System.out.println(value.getSourceTable() + "--------------");
                        String key = getKey(value.getSourceTable(), value.getSourceType());
                        broadcastState.put(key,value);
                    }
                });
    }

    /**
     * @description: create table on phoenix
     * @author: zenghaifeng
     * @date: 2023/4/21 18:43
     * @param configTable
     * @return SingleOutputStreamOperator<TableProcess>
     **/
    private SingleOutputStreamOperator<TableProcess> createPhoenixTable(
            SingleOutputStreamOperator<TableProcess> configTable){
        return configTable.process(new ProcessFunction<TableProcess, TableProcess>() {
            private Connection connection;
            // connect to phoenix
            @Override
            public void open(Configuration parameters) {
                connection = JdbcUtil.getPhoenixConnection();
            }

            // close connect
            @Override
            public void close() {
                JdbcUtil.closeConnection(connection);
            }

            @Override
            public void processElement(TableProcess value,
                                       ProcessFunction<TableProcess, TableProcess>.Context ctx,
                                       Collector<TableProcess> out) throws Exception {
                String op = value.getOp();
                if ("r".equals(op) || "c".equals(op)){
                    createTable(value);
                }else if ("d".equals(op)){
                    dropTable(value);
                }else {
                    dropTable(value);
                    createTable(value);
                }
                out.collect(value);
            }
            /*
                create table if not exist tableName(
                    id varchar,
                    name varchar,
                    CONSTRAINT my_pk PRIMARY KEY (id)
                )salt_buckets = 4
             */
            private void createTable(TableProcess tableProcess) throws SQLException {
                StringBuilder createTableSql = new StringBuilder();
                createTableSql.append("create table if not exists ");
                createTableSql.append(tableProcess.getSinkTable());
                createTableSql.append("(");
                createTableSql.append(tableProcess.getSinkColumns().replaceAll("[^,]+","$0 varchar"));
                createTableSql.append(", constraint pk primary key(");
                createTableSql.append(tableProcess.getSinkPk() == null ? "id" : tableProcess.getSinkPk());
                createTableSql.append(")) ");
                createTableSql.append(tableProcess.getSinkExtend() == null ? "" : tableProcess.getSinkExtend());

                executeDDLSQL(createTableSql.toString());

            }
            private void dropTable(TableProcess tableProcess) throws SQLException {
                String dropTableSql = " drop table if exists " + tableProcess.getSinkTable();

                executeDDLSQL(dropTableSql);
            }

            private void executeDDLSQL(String sql) throws SQLException {
                PreparedStatement preparedStatement = connection.prepareStatement(sql);
                preparedStatement.execute();
                preparedStatement.close();
            }
        });

    }

    /**
     * @description: via flink cdc read config table
     * @author: zenghaifeng
     * @date: 2023/4/21 18:45
     * @param env
     * @return SingleOutputStreamOperator<TableProcess>
     **/
    private SingleOutputStreamOperator<TableProcess> readTableProcess(
            StreamExecutionEnvironment env) {
        Properties jdbcProperties = new Properties();
        jdbcProperties.setProperty("useSSL","false");
        MySqlSource<String> mysqlSource = MySqlSource
                .<String>builder()
                .hostname(Constant.MYSQL_HOST)
                .port(Constant.MYSQL_PORT)
                .databaseList("gmall2023_config")
                .tableList("gmall2023_config.table_process")
                .username("root")
                .password("aaaaaa")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .jdbcProperties(jdbcProperties)
                // 第一次启动先读取所有的数据，然后在监控binlog实时变化
                .startupOptions(StartupOptions.initial())
                .build();
        return env.fromSource(mysqlSource, WatermarkStrategy.noWatermarks(),"mysqlSource")
                /*
                 op: mysql cdc 采集到的操作状态
                 c => insert before: null            after: <data>
                 r => read   before: null            after: <data>
                 u => update before: <old data>      after: <new data>
                 d => delete before: <delete data>   after: null
                 */
                .map((MapFunction<String, TableProcess>) value -> {
                    JSONObject jsonObject = JSON.parseObject(value);
                    String op = jsonObject.getString("op");
                    TableProcess tp;
                    if ("d".equals(op)){
                        tp = jsonObject.getObject("before", TableProcess.class);
                    }else {
                        tp = jsonObject.getObject("after", TableProcess.class);
                    }
                    // 手动 set op 字段
                    tp.setOp(op);
                    return tp;
                })
                // filter dim
                .filter(tp -> "dim".equals(tp.getSinkType()));
    }

    /**
     * @description: 清洗数据
     * @author: zenghaifeng
     * @date: 2023/4/21 18:45
     * @param ds
     * @return SingleOutputStreamOperator<JSONObject>
     **/
    private SingleOutputStreamOperator<JSONObject> etl(
            DataStreamSource<String> ds) {
        return ds.filter((FilterFunction<String>) value -> {
            try {
                JSONObject obj = JSON.parseObject(value);
                String type = obj.getString("type");
                return "gmall2023".equals(obj.getString("database")) &&
                        ("insert".equals(type) || "update".equals(type) || "bootstrap-insert".equals(type)) &&
                        null != obj.getJSONObject("data");
            } catch (Exception e) {
                log.warn("数据格式有误,不是正确的 json 数据: " + "value");
                return false;
            }
        }).map(json -> JSON.parseObject(json.replace("bootstrap-","")));
    }
}
